package com.leilei;

import cn.hutool.core.util.RandomUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.Random;

/**
 * @author lei
 * @version 1.0
 * @desc flink 有状态计算 状态过期设置（ttl）
 * @date 2021-03-19 15:39
 */
public class Flink_State_3_TTL {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(4);
        final DataStreamSource<Location> locationSource = env.addSource(new LocationSource());
        final WindowedStream<Location, Integer, GlobalWindow> windowedStream = locationSource.keyBy(Location::getVehicleId)
                .countWindow(1);
        windowedStream.apply(new SpeedAlarmWindow()).print();
        env.execute("state-ttl");

    }


    public static class SpeedAlarmWindow extends RichWindowFunction<Location, String, Integer, GlobalWindow> {
        MapState<String, Location> locationState;

        @Override
        public void apply(Integer integer, GlobalWindow window, Iterable<Location> locationList, Collector<String> out) throws Exception {
            for (Location location : locationList) {
                final String key = location.getVehicleId().toString();
                final Location preLocation = locationState.get(key);
                if (preLocation == null) {
                    if (location.getGpsSpeed() > location.getLimitSpeed()) {
                        locationState.put(key, location);
                        out.collect(location.toString() + "超速开始");
                        return;
                    }
                } else {
                    if (location.getGpsSpeed() > location.getLimitSpeed()) {
                        locationState.put(key, location);
                        out.collect(location.toString() + "持续超速中" + ">> " +
                                "上一条超速数据为：" + "\n" + preLocation.toString());

                    } else {
                        locationState.remove(key);
                        out.collect(location.toString() + "超速结束");
                    }
                }
            }
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            StateTtlConfig ttlConfig = StateTtlConfig
                    // 状态有效时间
                    .newBuilder(Time.seconds(10))
                    //设置状态更新类型
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    // 已过期但还未被清理掉的状态如何处理
                    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                    // 过期对象的清理策略
                    .cleanupFullSnapshot()
                    .build();
            //MapState 状态管理配置
            MapStateDescriptor<String, Location> mapStateDescriptor = new MapStateDescriptor<>("locationState",
                    TypeInformation.of(String.class),
                    TypeInformation.of(Location.class));
            //启用状态存活时间设置
            mapStateDescriptor.enableTimeToLive(ttlConfig);
            this.locationState = getRuntimeContext().getMapState(mapStateDescriptor);
        }
    }

    public static class LocationSource implements SourceFunction<Location> {

        Boolean flag = true;

        @Override
        public void run(SourceContext<Location> ctx) throws Exception {
            Random random = new Random();
            while (flag) {
                int vehicleId = random.nextInt(2) + 1;
                Location location = Location.builder()
                        .vehicleId(vehicleId)
                        .plate("川A000" + vehicleId)
                        .color("绿")
                        .date(Integer.parseInt(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE)))
                        .gpsSpeed(RandomUtil.randomInt(88, 100))
                        .limitSpeed(RandomUtil.randomInt(88, 95))
                        .devTime(System.currentTimeMillis())
                        .build();
                ctx.collect(location);
                Thread.sleep(RandomUtil.randomInt(5,15)*1000);

            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}
